{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Agenda"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Distributed mean and median algorithm\n",
    "- Sorter and collector implementation\n",
    "  - [muduo-protorpc/examples/median](https://github.com/chenshuo/muduo-protorpc/tree/cpp11/examples/median)\n",
    "- Implement an asynchronous RPC framework in 10 slides"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Problem Definition"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```\n",
    "多机数据处理。有 10 台机器, 每台机器上保存着 10亿个 64-bit 整数 (不一定刚好 10 亿个, 可能有上下几千万的浮动), 一共约 100 亿个整数（其实一共也就 80GB 数据, 不算大, 选这个量级是考虑了VPS 虚拟机的容量, 便于实验）。编程求出:\n",
    "\n",
    "1. 这些数的平均数。\n",
    "2. 这些数的中位数。\n",
    "3. 出现次数最多的 100 万个数。\n",
    "*4. （附加题）对这100 亿个整数排序, 结果顺序存放到这 10 台机器上。\n",
    "*5. （附加健壮性要求）你的程序应该能正确应对输入数据的各种分布(均匀、正态、Zipf)。\n",
    "*6. （附加伸缩性要求）你的程序应该能平滑扩展到更多的机器, 支持更大的数据量。比如 20 台机器、一共 200亿个整数, 或者 50 台机器、一共 500 亿个整数。\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Distributed Algorithm"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Mean: $\\text{Mean} = \\frac{\\sum_i \\text{sum}_i}{\\sum_i \\text{count}_i}$\n",
    "- *k-th* element algorithm by **bisection**  \n",
    "  - [https://en.wikipedia.org/wiki/Bisection_method](https://en.wikipedia.org/wiki/Bisection_method)  \n",
    "  - For a **guess**, find number of elements that is **smaller** than *guess*, and number of elements that **equals** to *guess*  \n",
    "  - Keep improving *guess* until:$\\text{smaller} < k \\quad \\text{and} \\quad \\text{smaller} + \\text{same} \\geq k$\n",
    "- **Median** is the element at: $\\left\\lfloor \\frac{N}{2} \\right\\rfloor$\n",
    "  Examples:  \n",
    "  - $[1, 2, 3], \\ \\text{median} = 2$  \n",
    "  - $[1, 2, 3, 4], \\ \\text{median} = 3 \\ (\\text{not } 2.5)$"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "平均数\n",
    "- 计算公式:$\\operatorname{Mean} = \\sum \\operatorname{sum}_i / \\sum count_i$\n",
    "- 实现方法: 每台机器计算本地sum和count，collector汇总后做除法"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "中位数 \n",
    "- k-th元素算法 \n",
    "  - 算法原理: 基于二分法(bisection method)的分布式实现\n",
    "  - 计算步骤:\n",
    "    - 猜测一个数值guess\n",
    "    - 统计所有机器上比guess小的元素数量smaller\n",
    "    - 统计所有机器上与guess相等的元素数量same\n",
    "    - 根据smaller和same调整guess范围\n",
    "    - 重复直到满足$smaller < k$且 $smaller + same \\geq k$\n",
    "  - 中位数定义: 取第$\\lfloor n/2 \\rfloor$个元素(非数学标准定义)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 算法实现"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## kth.h"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo-protorpc/blob/cpp11/examples/median/kth.h\n",
    "\n",
    "#ifndef MUDUO_PROTORPC2_EXAMPLES_MEDIAN_KTH_H\n",
    "#define MUDUO_PROTORPC2_EXAMPLES_MEDIAN_KTH_H\n",
    "\n",
    "#include <muduo/base/Logging.h>\n",
    "\n",
    "#include <functional>\n",
    "\n",
    "// https://gist.github.com/3703911\n",
    "// array: [-1, 0, 1, 2, 3, 8 ]\n",
    "// K-th : [ 1, 2, 3, 4, 5, 6 ]\n",
    "inline\n",
    "std::pair<int64_t, bool> getKth(std::function<void(int64_t, int64_t*, int64_t*)> search,\n",
    "                                const int64_t k,\n",
    "                                const int64_t count, int64_t min, int64_t max)\n",
    "{\n",
    "  int steps = 0;\n",
    "  int64_t guess = max;\n",
    "  bool succeed = false;\n",
    "  while (min <= max)\n",
    "  {\n",
    "    int64_t smaller = 0;\n",
    "    int64_t same = 0;\n",
    "    search(guess, &smaller, &same);\n",
    "    LOG_INFO << \"guess = \" << guess\n",
    "             << \", smaller = \" << smaller\n",
    "             << \", same = \" << same\n",
    "             << \", min = \" << min\n",
    "             << \", max = \" << max;\n",
    "\n",
    "    if (++steps > 100) {\n",
    "      LOG_ERROR << \"Algorithm failed, too many steps\";\n",
    "      break;\n",
    "    }\n",
    "\n",
    "    if (smaller < k && smaller + same >= k) {\n",
    "      succeed = true;\n",
    "      break;\n",
    "    }\n",
    "\n",
    "    if (smaller + same < k) {\n",
    "      min = guess + 1;\n",
    "    } else if (smaller >= k) {\n",
    "      max = guess;\n",
    "    } else {\n",
    "      LOG_ERROR << \"Algorithm failed, no improvement\";\n",
    "      break;\n",
    "    }\n",
    "    guess = min + static_cast<uint64_t>(max - min)/2;\n",
    "    assert(min <= guess && guess <= max);\n",
    "  }\n",
    "\n",
    "  std::pair<int64_t, bool> result(guess, succeed);\n",
    "  return result;\n",
    "}\n",
    "\n",
    "#endif // MUDUO_PROTORPC2_EXAMPLES_MEDIAN_KTH_H"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## kth_unittest.cc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo-protorpc/blob/cpp11/examples/median/kth_unittest.cc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Service definition"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `func Query() (count, min, max, sum int64)`\n",
    "- `func Search(guess int64) (smaller, same int64)`\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- https://github.com/chenshuo/muduo-protorpc/blob/cpp11/examples/median/median.proto"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Query接口:\n",
    "  - 返回: (count, min, max, sum)\n",
    "  - 用途: 获取机器数据的基本统计信息\n",
    "- Search接口:\n",
    "  - 输入: guess\n",
    "  - 返回: (smaller, same)\n",
    "  - 用途: 实现k-th元素算法"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Sorter and Collector"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![11.1](./images/11.1.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- https://github.com/chenshuo/muduo-protorpc/tree/cpp11/examples/median"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Sorter(服务端):\n",
    "  - 存储排序后的数据数组\n",
    "  - 实现Query和Search接口\n",
    "- Collector(客户端):\n",
    "  - 通过RPC调用多个Sorter\n",
    "  - 汇总结果计算平均数和中位数\n",
    "- 通信机制:\n",
    "  - 使用protobuf定义RPC接口\n",
    "  - Sorter_Stub封装RPC调用细节"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 代码实现及运行实例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## sorter.cc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo-protorpc/blob/cpp11/examples/median/sorter.cc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## collector.cc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo-protorpc/blob/cpp11/examples/median/collector.cc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 运行"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 多台机器求中位数"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "bin/median_collector 127.0.0.1:5555 127.0.0.1:6666 127.0.0.1:7777\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "bin/median_sorter # 默认5555\n",
    "```\n",
    "\n",
    "```shell\n",
    "bin/median_sorter 6666\n",
    "```\n",
    "\n",
    "```shell\n",
    "bin/median_sorter 7777\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 验证程序中位数"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "bin/median_collector 127.0.0.1:5555\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "bin/median_sorter # 默认5555\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# TCP Incast Problem"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- http://www.pdl.cmu.edu/Incast/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![11.2](./images/11.2.jpg)\n",
    "\n",
    "-  A simple cluster-based storage environment with one client requesting data from multiple servers through synchronized reads."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 问题本质：\n",
    "  - 多服务器同时响应导致客户端侧拥塞\n",
    "  - 典型场景：分布式查询/MapReduce\n",
    "- 表现特征：\n",
    "  - 服务端快速完成但客户端出现丢包\n",
    "  - Linux默认200ms重传延迟\n",
    "- 优化方向：\n",
    "  - 调整TCP缓冲区大小\n",
    "  - 实现响应速率控制\n",
    "  - 采用拥塞避免算法"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Implement a RPC framework"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- https://github.com/chenshuo/muduo/blob/master/muduo/net/protorpc/rpc.proto"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "properties"
    }
   },
   "outputs": [],
   "source": [
    "package muduo.net;\n",
    "// option go_package = \"muduorpc\";\n",
    "option java_package = \"com.chenshuo.muduo.protorpc\";\n",
    "option java_outer_classname = \"RpcProto\";\n",
    "\n",
    "enum MessageType\n",
    "{\n",
    "  REQUEST = 1;\n",
    "  RESPONSE = 2;\n",
    "  ERROR = 3; // not used\n",
    "}\n",
    "\n",
    "enum ErrorCode\n",
    "{\n",
    "  NO_ERROR = 0;\n",
    "  WRONG_PROTO = 1;\n",
    "  NO_SERVICE = 2;\n",
    "  NO_METHOD = 3;\n",
    "  INVALID_REQUEST = 4;\n",
    "  INVALID_RESPONSE = 5;\n",
    "  TIMEOUT = 6;\n",
    "}\n",
    "\n",
    "message RpcMessage\n",
    "{\n",
    "  required MessageType type = 1;\n",
    "  required fixed64 id = 2;\n",
    "\n",
    "  optional string service = 3;\n",
    "  optional string method = 4;\n",
    "  optional bytes request = 5;\n",
    "\n",
    "  optional bytes response = 6;\n",
    "\n",
    "  optional ErrorCode error = 7;\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- RPC = message passing + method dispatch"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 核心组成：RPC = 消息传递 + 方法分发\n",
    "- 消息字段：\n",
    "  - type：标识请求或响应类型（REQUEST/RESPONSE）\n",
    "  - id：64位固定长度ID，用于异步调用匹配\n",
    "  - service：可选服务名（对应类名）\n",
    "  - method：可选方法名（对应函数名）\n",
    "  - request/response：二进制数据载荷\n",
    "  - error：错误码字段\n",
    "- 设计特点：采用Protobuf序列化方案，简化数据传递实现"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Service definition"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "properties"
    }
   },
   "outputs": [],
   "source": [
    "package sudoku;\n",
    "option cc_generic_services = true;\n",
    "\n",
    "message SudokuRequest {\n",
    "    required string checkerboard = 1;\n",
    "}\n",
    "\n",
    "message SudokuResponse {\n",
    "    optional bool solved  = 1 [default = false];\n",
    "    optional string checkerboard = 2;\n",
    "}\n",
    "\n",
    "service SudokuService {\n",
    "    rpc Solve (SudokuRequest) returns (SudokuResponse);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 服务示例：数独求解服务\n",
    "  - `SudokuRequest`：包含81字符棋盘（空白处填0）\n",
    "  - `SudokuResponse`：包含求解状态和结果棋盘\n",
    "- 服务定义：\n",
    "- 实现要点：通过`cc_generic_services = true`选项启用代码生成"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Generated Service Interface"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "class SudokuService_Stub;\n",
    "\n",
    "class SudokuService : public ::google::protobuf::Service {\n",
    "public:\n",
    "    typedef SudokuService_Stub Stub;\n",
    "\n",
    "    // Implemented by SudokuService_Stub and SudokuServiceImpl\n",
    "    virtual void Solve(RpcController* controller,\n",
    "                       const SudokuRequest* request,\n",
    "                       SudokuResponse* response,\n",
    "                       Closure* done) = 0;\n",
    "    // Closure is like std::function<void()>\n",
    "\n",
    "    static const ::google::protobuf::ServiceDescriptor* descriptor();\n",
    "    const ServiceDescriptor* GetDescriptor() override { return descriptor(); }\n",
    "\n",
    "    // Called by server runtime to dispatch request to its method\n",
    "    void CallMethod(const MethodDescriptor* method,\n",
    "                    RpcController* controller,\n",
    "                    const Message* request,\n",
    "                    Message* response,\n",
    "                    Closure* done) override;\n",
    "};"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 核心类结构：\n",
    "  - `SudokuService`：抽象基类，继承自`google::protobuf::Service`\n",
    "  - `SudokuService_Stub`：客户端存根类\n",
    "  - `SudokuServiceImpl`：服务端实现类\n",
    "- 关键方法：\n",
    "  - `Solve()`：纯虚函数，需子类实现具体逻辑\n",
    "  - `CallMethod()`：服务端用于方法分发的入口\n",
    "  - `GetDescriptor()`：获取服务元数据（反射机制）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# SudokuService::CallMethod"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "void SudokuService::CallMethod(const MethodDescriptor* method,\n",
    "                               RpcController* controller,\n",
    "                               const Message* request,   // doesn't take ownership\n",
    "                               Message* response,        // doesn't take ownership\n",
    "                               Closure* done) {\n",
    "    GOOGLE_DCHECK_EQ(method->service(), SudokuService_descriptor_);\n",
    "    switch (method->index()) {  // vtbl\n",
    "        case 0:\n",
    "            Solve(controller,\n",
    "                  down_cast<const SudokuRequest*>(request),\n",
    "                  down_cast<SudokuResponse*>(response),\n",
    "                  done);\n",
    "            break;\n",
    "        default:\n",
    "            GOOGLE_LOG(FATAL) << \"Bad method index; this should never happen.\";\n",
    "            break;\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 分发机制：\n",
    "  - 通过`method->index()`进行`switch-case`分发\n",
    "  - 使用`down_cast`进行安全类型转换（debug时`dynamic_cast`，release时`static_cast`）\n",
    "- 参数处理：\n",
    "  - `request/response`为裸指针但不获取所有权\n",
    "  - `controller`参数在本框架中未使用（简化实现）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Implement SudokuServer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "class SudokuServiceImpl : public SudokuService {\n",
    "public:\n",
    "    void Solve(::google::protobuf::RpcController* controller,\n",
    "               const ::sudoku::SudokuRequest* request,   // doesn’t take ownership\n",
    "               ::sudoku::SudokuResponse* response,       // doesn’t take ownership\n",
    "               ::google::protobuf::Closure* done) override\n",
    "    {\n",
    "        LOG_INFO << \"SudokuServiceImpl::Solve\";\n",
    "        response->set_solved(true);\n",
    "        response->set_checkerboard(\"1234567\");\n",
    "        done->Run();\n",
    "    }\n",
    "};"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 实现要点：\n",
    "  - 继承自生成的`SudokuService`接口类\n",
    "  - 必须实现`Solve()`纯虚函数\n",
    "  - 同步实现模式：直接设置r`esponse`并调用`done->Run()`\n",
    "- 所有权约定：\n",
    "  - 框架维护`request/response`生命周期\n",
    "  - 实现者不应`delete`传入的指针参数"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# RpcServer pseudo code"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "class RpcServer {\n",
    "    std::map<std::string, Service*> services_;\n",
    "\n",
    "    void onRpcMessage(const TcpConnection& conn, const RpcMessage& message) {\n",
    "        Service* service = services_[message.service()];\n",
    "        const ServiceDescriptor* desc = service->GetDescriptor();\n",
    "        const MethodDescriptor* method = desc->FindMethodByName(message.method());\n",
    "        Message* request = Deserialize(message.request());\n",
    "        Message* response = service->GetResponsePrototype(method).New();\n",
    "        int64_t id = message.id();\n",
    "\n",
    "        service->CallMethod(method, NULL /* RpcController */, request, response,\n",
    "            [=] {\n",
    "                RpcMessage message;\n",
    "                message.set_type(RESPONSE);\n",
    "                message.set_id(id);\n",
    "                message.set_response(response->SerializeAsString());\n",
    "                codec_.send(conn, message);   // RpcCodec codec_\n",
    "                delete request;\n",
    "                delete response;\n",
    "            });\n",
    "    }\n",
    "};"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 核心流程：\n",
    "  - 通过`service`名查找注册的服务实例\n",
    "  - 通过反射获取方法描述符\n",
    "  - 反序列化请求数据\n",
    "  - 创建空响应对象\n",
    "  - 调用目标方法\n",
    "- 回调处理：\n",
    "  - 使用lambda封装响应构造逻辑\n",
    "  - 自动处理序列化和资源释放\n",
    "- 重要细节：\n",
    "  - 使用`ServiceDescriptor`进行反射查询\n",
    "  - 通过`GetResponsePrototype`创建类型正确的响应对象"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Server call trace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "TcpConnection::onMessage()\n",
    "  RpcCodec::onMessage()             // decodes RpcMessage\n",
    "    RpcServer::onRpcMessage()\n",
    "      // parse request, prepare empty response\n",
    "      // find service, then service->CallMethod()\n",
    "        SudokuService::CallMethod() override\n",
    "          // dispatch, call SudokuService::Solve()\n",
    "            SudokuServiceImpl::Solve()\n",
    "\n",
    "done()\n",
    "  RpcServer::<lambda>()\n",
    "    // Construct and send RpcMessage response\n",
    "    // free resource"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 完整调用链：\n",
    "  - TCP层接收原始数据\n",
    "  - `RpcCodec`解码为`RpcMessage`\n",
    "  - `RpcServer`路由到具体`Service`\n",
    "  - 通过`CallMethod`分发到实现类\n",
    "  - 业务逻辑处理完成触发回调\n",
    "- 响应路径：\n",
    "  - 在lambda中构造响应消息\n",
    "  - 通过`codec`发送序列化结果\n",
    "  - 最后释放`request/response`资源"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Generated Client Stub"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "class SudokuService_Stub : public SudokuService {\n",
    "public:\n",
    "    SudokuService_Stub(::google::protobuf::RpcChannel* channel);  // set channel_\n",
    "\n",
    "    void Solve(::google::protobuf::RpcController* controller,\n",
    "               const ::sudoku::SudokuRequest* request,   // doesn’t take ownership\n",
    "               ::sudoku::SudokuResponse* response,       // takes ownership!!!\n",
    "               ::google::protobuf::Closure* done) override\n",
    "    {\n",
    "        channel_->CallMethod(descriptor()->method(0),\n",
    "                             controller, request, response, done);\n",
    "    }\n",
    "\n",
    "private:\n",
    "    ::google::protobuf::RpcChannel* channel_;\n",
    "};"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 自动生成机制：由protobuf编译器自动生成`SudokuService_Stub`类，继承自`SudokuService`基类\n",
    "- 核心方法：\n",
    "  - 构造函数：接收`RpcChannel*`参数用于通信\n",
    "  - `Solve`方法：通过`channel\\_->CallMethod()$`转发调用，参数包含：\n",
    "    - `controller`：RPC控制器\n",
    "    - `request`：调用方保留所有权（不需主动`delete`）\n",
    "    - `response`：存根取得所有权（必须new创建）\n",
    "    - `done`：异步回调闭包\n",
    "- 所有权规则：特别注意`response`指针的所有权转移，调用后禁止手动`delete`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Sudoku Client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "RpcChannel channel;\n",
    "TcpConnectionPtr conn;   // an established TCP connection\n",
    "conn->setMessageCallback(\n",
    "    ::bind(&RpcChannel::onMessage, &channel, _1, _2, _3));\n",
    "channel.setConnection(conn);\n",
    "SudokuService::Stub stub(&channel);\n",
    "\n",
    "SudokuRequest request;  \n",
    "request.set_checkerboard(\"001030\");\n",
    "SudokuResponse* response = new SudokuResponse;\n",
    "\n",
    "stub.Solve(NULL /* RpcController */, &request, response, [response]() {\n",
    "    LOG_INFO << response->DebugString();\n",
    "});"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 组件关系：\n",
    "  - `RpcChannel` 与 `TcpConnection` 绑定，通过 `setMessageCallback` 设置消息处理\n",
    "  - 一个 `channel` 可复用服务多个 `service`\n",
    "- 调用流程\n",
    "  - 构造 `SudokuRequest` 并设置棋盘数据（如 `request.set_checkerboard(\"001030\")`）\n",
    "  - 必须用 `new` 创建 `SudokuResponse`（因所有权转移）\n",
    "  - 调用 `stub.Solve()` 时\n",
    "    - 可置空 `RpcController`\n",
    "    - 通过 lambda 捕获 `response` 打印结果（`response->DebugString()`）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# RpcChannel members"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "namespace muduo {\n",
    "namespace net {\n",
    "\n",
    "class RpcChannel : public ::google::protobuf::RpcChannel {\n",
    "public:\n",
    "    void CallMethod(...) override;   // inherited from RpcChannel\n",
    "\n",
    "private:\n",
    "    struct OutstandingCall {\n",
    "        ::google::protobuf::Message* response;\n",
    "        ::google::protobuf::Closure* done;\n",
    "    };\n",
    "\n",
    "    RpcCodec codec_;   // calls onRpcMessage()\n",
    "    TcpConnectionPtr conn_;\n",
    "    AtomicInt64 id_;\n",
    "    std::map<int64_t, OutstandingCall> outstandings_;\n",
    "};\n",
    "\n",
    "} // namespace net\n",
    "} // namespace muduo"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 核心组件\n",
    "  - `outstandings_`：记录未完成 RPC 的映射表（key 为递增 ID）  \n",
    "  - `codec_`：负责消息编解码  \n",
    "  - `conn_`：关联的 TCP 连接  \n",
    "- 数据结构\n",
    "  - `OutstandingCall` 保存响应消息和回调函数包  \n",
    "  - `AtomicInt64 id_` 保证线程安全 ID 生成  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# RpcChannel::CallMethod"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "void RpcChannel::CallMethod(const MethodDescriptor* method,\n",
    "                            RpcController* ctr,\n",
    "                            const Message* request,\n",
    "                            Message* response,\n",
    "                            Closure* done)\n",
    "{\n",
    "    RpcMessage message;\n",
    "    message.set_type(REQUEST);\n",
    "\n",
    "    const int64_t id = id_.incrementAndGet();\n",
    "    message.set_id(id);\n",
    "    message.set_service(method->service()->name());\n",
    "    message.set_method(method->name());\n",
    "    message.set_request(request->SerializeAsString());\n",
    "\n",
    "    OutstandingCall out = { response, done };   // we own response\n",
    "    outstandings_[id] = out;\n",
    "\n",
    "    codec_.send(conn_, message);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 关键步骤：\n",
    "  - 构造 RpcMessage 并设置：  \n",
    "    - `type = REQUEST`  \n",
    "    - 自增 ID（`id_.incrementAndGet()`）  \n",
    "    - 服务名与方法名  \n",
    "    - 序列化后的请求数据  \n",
    "  - 必须先存入 `outstandings_` 再发送（避免竞态条件）  \n",
    "  - 通过 `codec_.send()` 发送消息  \n",
    "\n",
    "- 线程安全：\n",
    "  - 若先发送后存储，可能因线程切换导致响应丢失  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# RpcChannel::onRpcMessage"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "void RpcChannel::onRpcMessage(const RpcMessage& message) {\n",
    "    assert(message.type() == RESPONSE);\n",
    "\n",
    "    // omit all error handling\n",
    "    auto it = outstandings_.find(message.id());\n",
    "    OutstandingCall out = it->second;\n",
    "    outstandings_.erase(it);\n",
    "\n",
    "    out.response->ParseFromString(message.response());\n",
    "    out.done->Run();\n",
    "    delete out.response;\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 处理流程：\n",
    "  - 验证消息类型为 `RESPONSE`  \n",
    "  - 根据 ID 从 `outstandings_` 查找并移除记录  \n",
    "  - 反序列化响应数据（`response->ParseFromString()`）  \n",
    "  - 执行回调（`done->Run()`）  \n",
    "  - 必须删除 `response` 防止内存泄漏  \n",
    "\n",
    "- 错误处理：\n",
    "  - 未找到对应 ID 时应丢弃异常响应  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Client call trace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "Client calls: SudokuService::Stub::Solve(Closure* done)\n",
    "  RpcChannel::CallMethod()`\n",
    "    // `done` callback to outstanding RPC map  \n",
    "    // Add Construct and send `RpcMessage` request  \n",
    "\n",
    "TcpConnection::onMessage()\n",
    "  RpcCodec::onMessage()    // decodes RpcMessage  \n",
    "    RpcChannel::onRpcMessage()\n",
    "    // Find `done` callback from outstanding RPC map  \n",
    "      done->Run()\n",
    "    // Free resource"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 完整调用链：\n",
    "  - 客户端调用 `SudokuService::Stub::Solve()`\n",
    "  - `RpcChannel::CallMethod()` 构造并发送请求  \n",
    "  - 服务端响应经  \n",
    "    `TcpConnection -> RpcCodec -> RpcChannel`  \n",
    "    解码  \n",
    "  - 触发 `onRpcMessage` 执行回调并清理资源  \n",
    "\n",
    "- 关键顺序：\n",
    "  - 注册回调必须先于发送请求"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Asynchronous Server"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Clients are always asynchronous, because we could be in EventLoop thread,  \n",
    "  if we block, no chance to read response  \n",
    "- Sudoku Server is synchronous  \n",
    "- DNS resolver is an example of asynchronous server  \n",
    "  - [muduo/examples/protobuf/resolver ](https://github.com/chenshuo/muduo/tree/master/examples/protobuf/resolver)\n",
    "  - Use **c-ares** for asynchronous DNS resolution  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 客户端异步必要性：客户端必须采用异步设计，因为可能运行在`EventLoop`线程中。若采用同步阻塞方式，将无法及时处理网络IO事件（如读取响应），导致程序失去响应能力。\n",
    "- 服务端同步可行性：服务端可以同步执行CPU密集型计算（如Sudoku求解），但严禁同步等待IO操作（如DNS查询），否则会阻塞IO线程，违反muduo事件驱动编程原则。\n",
    "- 异步服务端示例：DNS解析服务是典型异步实现案例，使用c-ares库进行异步解析（代码参考muduo/examples/protobuf/resolver）。当服务端需要依赖其他服务时，应采用回调机制而非线程阻塞。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Why and why not muduo-protorpc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> C++11后, 可能有变"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Better **ownership** with smart pointers, not **raw pointers**  \n",
    "  - Should be **unique_ptr**, but is **shared_ptr**  \n",
    "- **std::function** vs. **Closure**  \n",
    "- Doesn’t implement **timeout**  \n",
    "- Doesn’t implement RPC **cancel**  \n",
    "- Doesn’t have good **debug** support  \n",
    "- Use **gRPC** in your project  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "muduo-protorpc实现的优势\n",
    "\n",
    "- 所有权管理改进：采用智能指针（当时使用`shared_ptr`）替代Google原生的原始指针，避免手动管理所有权问题。现代实现应优先使用`unique_ptr`。\n",
    "- 回调机制优化：使用`std::function`替代Google自制的Closure基类，与muduo库整体风格更协调，提供更自然的C++编程接口。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "muduo-protorpc实现的工作量\n",
    "- 代码生成器开发：需要自行实现protobuf编译器插件，无法直接使用Google提供的代码生成器，显著增加开发工作量。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "muduo-protorpc未实现的功能\n",
    "\n",
    "关键功能缺失：\n",
    "- 超时控制（timeout）：无法自动释放长时间无响应的请求资源\n",
    "- 请求取消（cancel）：缺乏终止长时间运行请求的机制\n",
    "- 资源泄漏风险：服务端死锁时客户端持续累积请求可能导致内存耗尽"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "muduo-protorpc的调试功能\n",
    "- 监控能力不足：缺乏类似HTTP的实时监控界面，无法查看RPC延迟、QPS、吞吐量等关键指标，不利于生产环境问题诊断。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "muduo-protorpc与gRPC的对比\n",
    "\n",
    "生产环境建议：\n",
    "\n",
    "- muduo网络库本身适合生产环境\n",
    "- muduo-protorpc仅适合学习研究（约几百行代码）\n",
    "- 生产级RPC推荐使用gRPC，其优势包括：\n",
    "  - 跨语言支持（核心用C实现）\n",
    "  - 各语言绑定共享高效底层实现\n",
    "  - 避免多语言独立实现的兼容性问题"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "C++17",
   "language": "C++17",
   "name": "xcpp17"
  },
  "language_info": {
   "name": "C++17"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
